package cn.doitedu.day04.demo

import cn.doitedu.day01.utils.SparkUtil
import org.apache.spark.rdd.RDD

/**
 * @Date 22.4.1
 * @Created by HANGGE
 * @Description
 */
object Demo02_订单 {
  /**
   *       日志数据  (文件中)
   * 订单ID  商品类别  商品金额  商品ID
   * 1001,办公用品,900,1
   * 1002,生活用品,88,2
   * 1003,服装,699,3
   * 1004,办公用品,800,4
   * 1005,补品,9.9,6
   * 1006,服装,899,5
   * 1007,补品,99.88,7
   *
   *  商品信息 在数据库中
   * create table tb_goods(
   * pid int ,
   * name varchar(50),
   * describe  varchar(50)
   * ) ;
   * insert into tb_goods values
   * 1,"cherry机械键盘","just so so"),
   * 2,"雕牌卫生纸","厚而不硬"),
   * 3,"梦之蓝男装-格子衬衫","帅无边"),
   * 4,"奥妙打印机","你猜不到的结果"),
   * 5,"劲霸男装-短裤",""),
   * 6,"九全大补丸","就差那么一点点"),
   * 7,"多易牌黑枸杞","学大数, 不费劲") ;
   *
   * 1) 每种类别下的商品总金额
   * 2) 关联MySQL库中的商品名和描述信息 , 将结果存储到HDFS上
   *            - map算子
   *            - mapPartitions算子
   *            - MySQL数据构建RDD join
   *            - 广播变量
   */

  def main(args: Array[String]): Unit = {

    val sc = SparkUtil.getSc
    // 加载订单数据
    val data = sc.textFile("data/order.txt")
    // 处理每行订单
    val rdd1 = data.map(line => {
      val arr = line.split(",")
      //1005,补品,9.9,6
      (arr(0), arr(1), arr(2).toDouble, arr(3).toInt)
    })
    // 每种类别的商品总金额
    // 类别分组
    val rdd2: RDD[(String, Iterable[(String, String, Double, Int)])] = rdd1.groupBy(_._2)
  // 统计组内的数据
    val res = rdd2.map(tp=>{
      val category  = tp._1
      val ammount  = tp._2.map(_._3).toList.sum
      (category , ammount)
    })

    /**
     * (补品,109.78)
     * (生活用品,88.0)
     * (服装,1598.0)
     * (办公用品,1700.0)
     */

  }

}
